package com.mw.job.disruptor;

import com.lmax.disruptor.RingBuffer;
import com.mw.job.bean.AppProperties;
import com.mw.job.bean.DataBean;
import com.mw.job.util.HelpMe;
import com.mw.job.zookeeper.CuratorClient;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * Created by mengdexuan on 2017/9/29 16:15.
 */
@Component
public class DisruptorProducer implements Runnable {

	private Logger log = LoggerFactory.getLogger(getClass());

//	private static final String lockPath = "/MW-JOB/lock";

	private RingBuffer<DataBean> ringBuffer;

	@Autowired
	private JdbcTemplate jdbcTemplate;

	@Autowired
	AppProperties appProperties;

	private InterProcessMutex lock = null;

	private boolean isStop = false;//停止生产标识

	@Override
	public void run() {

		if (isStop) return;

		if (lock==null){
			String lockPath = "/MW-JOB/"+appProperties.getKey()+"/lock";
			lock = new InterProcessMutex(CuratorClient.curatorClient(), lockPath);
			log.info("创建分布式锁对象完成！");
		}

		try {
			boolean flag = lock.acquire(300, TimeUnit.MILLISECONDS);//获取锁

			if (flag){
				List<Map<String, Object>> mapList;

				try {
					mapList = jdbcTemplate.queryForList("select id,name,status from test where status = 1 limit 1");
					if (HelpMe.isNotNull(mapList)) {
						for (Map<String, Object> map : mapList) {
							jdbcTemplate.update("update test set status = 2 where id = " + HelpMe.getIntFromMap("id", map));
						}
					}
				}finally {
					lock.release();//数据状态改变后，尽早的释放锁，利于其它进程获取到锁
				}


				try {
					if (HelpMe.isNotNull(mapList)) {
						for (Map<String, Object> map : mapList) {
							publishData(map);//若环形队列满了，则生产线程会被阻塞
						}
						Thread.sleep(1000);//模拟生产用时
					}
				} catch (Exception e) {
					e.printStackTrace();
				}
			}

		} catch (Exception e) {
			e.printStackTrace();
		}

	}

	public void publishData(Map<String, Object> map) {
		long sequence = ringBuffer.next();
		DataBean event = null;
		try {
			event = ringBuffer.get(sequence);
			event.setId(HelpMe.getIntFromMap("id", map));
			event.setStatus(HelpMe.getIntFromMap("status", map));
			event.setName(HelpMe.getStrFromMap("name", map));
		} finally {
			ringBuffer.publish(sequence);//发布数据到环形队列中
			log.info("生产数据(P) --> " + event);
		}
	}

	public boolean isStop() {
		return isStop;
	}

	public void setStop(boolean stop) {
		isStop = stop;
	}

	public void setRingBuffer(RingBuffer<DataBean> ringBuffer) {
		this.ringBuffer = ringBuffer;
	}
}
